package com.lmc.log.function;


import com.lmc.log.bean.TrafficFlowType;
import net.lmc.util.DateTimeUtil;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class FlowTypeReportFunction implements WindowFunction<TrafficFlowType, String, String, TimeWindow> {

    @Override
    public void apply(String s, TimeWindow window, Iterable<TrafficFlowType> input, Collector<String> out) throws Exception {

        // 获取窗口开始和结束时间
        String windowStartTime = DateTimeUtil.convertLongToString(window.getStart(), DateTimeUtil.DATE_TIME_FORMAT);
        String windowEndTime = DateTimeUtil.convertLongToString(window.getEnd(), DateTimeUtil.DATE_TIME_FORMAT);

        // 对窗口中数据进行计算
        Long VehicleTypeCount = 0L;
        for (TrafficFlowType pageViewBean : input) {
            VehicleTypeCount += pageViewBean.getVehicleTypeCount();
        }

        // 拼接字符串
        String output = windowStartTime + "," +
                windowEndTime +"," +
                s + "," +
                VehicleTypeCount + "," +
                System.currentTimeMillis();

        out.collect(output);
//        System.out.println("输出结果 : " + output);
        
        
    }
}
